package com.hopu.shop.pro;

import com.alibaba.fastjson.JSONObject;
import com.hopu.bean.pro.Top10ParchasePro;
import com.hopu.util.HBaseUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SaveMode;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.util.List;
import java.util.Properties;

public class Top10ParchaseProAna {
    public static void main(String[] args) {

        SparkSession spark = SparkSession.builder()
                .config("spark.driver.allowMultipleContexts", "true")
                .config("spark.sql.crossJoin.enabled", "true")
                .appName("top10ParchaseProAna")
                .master("local").getOrCreate();
        JavaSparkContext context = new JavaSparkContext(spark.sparkContext());
        JavaRDD<String> rdd = context.textFile("D://word/user_session.log");
        JavaRDD<Top10ParchasePro> map = rdd.map(t -> {
            JSONObject json = (JSONObject) JSONObject.parse(t);
            String province = json.getString("province");
            String type = json.getString("event_type");
            return new Top10ParchasePro(province, type, 0);
        });

        JavaRDD<Top10ParchasePro> filter = map.filter(t -> "parchase".equals(t.getType()));
        Dataset<Row> df = spark.createDataFrame(filter, Top10ParchasePro.class);

        //各区域的总销量
        Dataset<Row> province = df.groupBy("province").count();
//        province.show();

        //排序，求前十区域的销量
        Dataset<Row> order = province.orderBy(province.col("count").desc()).limit(10);
        order.show();

        //保存到MySQL
        Properties pro = new Properties();
        pro.setProperty("driver", "com.mysql.jdbc.Driver");
        pro.setProperty("user", "root");
        pro.setProperty("password", "123456");
        order.write().mode(SaveMode.Append).jdbc("jdbc:mysql://192.168.136.200:3306/dataana?useUnicode=true&characterEncoding=UTF-8", "top10_pro_parchase", pro);


        //保存到HBase数据库
//        List<Row> rows = order.collectAsList();
//        String[] columns = {"province", "count"};
//        for (Row row: rows) {
//            String[] values = {row.get(0).toString(), row.get(1).toString()};
//            try {
//                HBaseUtils.putOneRowToHbase("pro_par", row.getString(0), "pro_par_top10", columns, values);
//            } catch (IOException e) {
//                e.printStackTrace();
//            }
//        }
    }
}
